/*
 * 在通信基类的基础上进一步实现了通信相关的子类
 * 如 MuduoBuffer(继承自BaseBuffer),LVProticol(继承自BaseProtocol)
 */
#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpClient.h> // muduo相关库，编译时加上 -lmuduo_net -lmuduo_base
#include <mutex>
#include <unordered_map>
#include "detail.hpp"
#include "fields.hpp"
#include "abstract.hpp"
#include "message.hpp"

namespace jsonrpc
{
    // MuduoBuffer继承自BaseBuffer
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *p_buf) : bufPtr(p_buf) {}
        virtual size_t readableSize() override
        {
            return bufPtr->readableBytes();
        }
        // 查看缓冲区中的下一个32位整数，但不移除它
        virtual int32_t peekInt32() override
        {
            // 注意: 在muduo的实现中，从缓冲区中读取4字节整数时，会进行字节序转换
            return bufPtr->peekInt32();
        }
        // 移除缓冲区中的下一个32位整数
        virtual void retrieveInt32() override
        {
            return bufPtr->retrieveInt32();
        }
        // 读取缓冲区中的下一个32位整数，并移除它
        virtual int32_t readInt32() override
        {
            return bufPtr->readInt32();
        }
        // 从缓冲区中读取指定长度的字符串，并移除这些数据
        virtual std::string retrieveAsString(size_t len) override
        {
            return bufPtr->retrieveAsString(len);
        }

    protected:
        muduo::net::Buffer *bufPtr;
    };
    // 用于生产MuduoBuffer的工厂类
    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    // LVProtocol继承自BaseProtocol,实现LV(length&val)协议
    // 消息字段
    // len(后面的总长度,不包括len本身) + mtype + idlen(id长度) + id + body(信息主体)
    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;

        // 检查是否可以处理该缓冲区
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if (buf->readableSize() < lengthFieldsSize)
            {
                return false;
            }
            int32_t totalLen = buf->peekInt32();
            if (buf->readableSize() < totalLen + lengthFieldsSize)
            {
                return false;
            }
            return true;
        }
        // 处理缓冲区中的数据，并返回消息
        //  len(后面的总长度,不包括len本身) + mtype + idlen(id长度) + id + body(信息主体)
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            int32_t totalLen = buf->readInt32();
            MType mtype = static_cast<MType>(buf->readInt32());
            int32_t idLen = buf->readInt32();
            int32_t bodyLen = totalLen - mtypeFieldsSize - idlenFieldsSize - idLen;
            std::string id = buf->retrieveAsString(idLen);
            std::string body = buf->retrieveAsString(bodyLen);
            msg = MessageFactory::create(mtype);
            if (msg.get() == nullptr)
            {
                LOG_ERROR("create message failed, error mtype");
                return false;
            }
            bool res = msg->deserialize(body);
            if (!res)
            {
                LOG_ERROR("deserialize message failed");
                return false;
            }
            msg->setRid(id);
            msg->setMessageType(mtype);
            return true;
        }
        // 序列化消息
        // len(后面的总长度,不包括len本身) + mtype + idlen(id长度) + id + body(信息主体)
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            std::string body = msg->serialize();
            std::string id = msg->getRid();
            uint32_t netMtype = htonl(static_cast<int32_t>(msg->getMessageType()));
            int32_t netIdLen = htonl(id.length());
            int32_t hostTotalLen = mtypeFieldsSize + idlenFieldsSize + id.size() + body.size();
            int32_t netTotalLen = htonl(hostTotalLen);
            std::string res;
            res.reserve(hostTotalLen);
            res.append(reinterpret_cast<const char *>(&netTotalLen), lengthFieldsSize);
            res.append(reinterpret_cast<const char *>(&netMtype), mtypeFieldsSize);
            res.append(reinterpret_cast<const char *>(&netIdLen), idlenFieldsSize);
            res.append(id);
            res.append(body);
            return res;
        }

    protected:
        const size_t lengthFieldsSize = 4;
        const size_t mtypeFieldsSize = 4;
        const size_t idlenFieldsSize = 4;
    };

    // 用于生产LVProtocol的工厂类
    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    // 继承自BaseConnection,实现了基于Muduo的网络连接
    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &protocol) : _protocol(protocol), _conn(conn) {}
        // 发送消息
        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string data = _protocol->serialize(msg);
            _conn->send(data);
        }
        // 关机
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        // 判断是否连接
        virtual bool connected() override
        {
            return _conn->connected();
        }

    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    // 用于生产MuduoConnection的工厂类
    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    // 继承自BaseServer,实现了基于Muduo的服务器
    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort),
                                _protocol(ProtocolFactory::create())
        {
        }
        // 启动服务器
        virtual void start() override
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.start();
            _baseloop.loop();
        }

    protected:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG_INFO("new connection from %s", conn->peerAddress().toIpPort().c_str());
                auto muduoConn = ConnectionFactory::create(conn, _protocol);
                {
                    // 使用互斥锁保证线程安全，出作用域自动释放锁
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduoConn));
                }
                if (connection_cb)
                {
                    connection_cb(muduoConn);
                }
            }
            else
            {
                LOG_INFO("connection %s disconnected", conn->peerAddress().toIpPort().c_str());
                BaseConnection::ptr muduoConn;
                {
                    // 使用互斥锁保证线程安全，出作用域自动释放锁
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduoConn = it->second;
                    _conns.erase(conn);
                }
                if (close_cb)
                {
                    close_cb(muduoConn);
                }
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG_DEBUG("Message come,begin to parse");
            auto baseBuffer = BufferFactory::create(buf);
            while (true)
            {
                // 判断是否可以解析
                if (!_protocol->canProcessed(baseBuffer))
                {
                    if (baseBuffer->readableSize() > maxDataSize)
                    {
                        LOG_ERROR("data too large, close connection");
                        conn->shutdown();
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                // 解析消息,即反序列化
                bool res = _protocol->onMessage(baseBuffer, msg);
                if (!res)
                {
                    LOG_ERROR("parse message error, close connection");
                    conn->shutdown();
                    return;
                }
                BaseConnection::ptr baseConn;
                {
                    // 使用互斥锁保证线程安全，出作用域自动释放锁
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    baseConn = it->second;
                }
                // 调用回调函数，将消息传递给上层进一步处理
                if (message_cb)
                {
                    message_cb(baseConn, msg);
                }
            }
        }

    protected:
        const int maxDataSize = 1024 * 1024;
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
        std::mutex _mutex; // 互斥锁
    };

    // MuduoServer工厂类
    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    // 继承自BaseClient，实现MuduoClient
    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &sip, int sport) : _protocol(ProtocolFactory::create()),
                                                         _baseloop(_loopThread.startLoop()),
                                                         _latch(1),
                                                         _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient") {}
        virtual void connect() override
        {
            LOG_DEBUG("set callback and connect to server");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            // 设置连接消息的回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this,
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 连接服务器
            _client.connect();
            _latch.wait();
            LOG_DEBUG("connect to server success");
        }

        virtual void shutdown() override
        {
            return _client.disconnect();
        }

        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (connected() == false)
            {
                LOG_ERROR("client is disconnected");
                return false;
            }
            _conn->send(msg);
            return true;
        }

        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }

        virtual bool connected() override
        {
            if(!_conn)
                return false;
            return _conn->connected();
        }

    protected:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if(conn->connected()){
                LOG_DEBUG("connect to server success");
                _latch.countDown();
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else{
                LOG_DEBUG("connect to server failed");
                _conn.reset();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG_DEBUG("receive message from server");
            auto baseBuffer = BufferFactory::create(buf);
            while(true){
                // 缓冲区数据不足
                if (_protocol->canProcessed(baseBuffer) == false) {
                    // 如果缓冲区数据超过最大限制，关闭连接
                    if (baseBuffer->readableSize() > maxDataSize) {
                        conn->shutdown();
                        LOG_ERROR("data size too large, close connection");
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(baseBuffer, msg);
                if (ret == false) {
                    conn->shutdown();
                    LOG_ERROR("data error, close connection");
                    return ;
                }
                if (message_cb) message_cb(_conn, msg);

            }
        }

        protected:
            const int maxDataSize = 1024 * 1024;
            BaseProtocol::ptr _protocol;
            BaseConnection::ptr _conn;
            muduo::CountDownLatch _latch;
            muduo::net::EventLoopThread _loopThread;
            muduo::net::EventLoop *_baseloop;
            muduo::net::TcpClient _client;
        };

        // MuduoClient工厂类
        class ClientFactory
        {
        public:
            template <typename... Args>
            static BaseClient::ptr create(Args &&...args)
            {
                return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
            }
        };
    }
